package com.study.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.lang.reflect.InvocationTargetException;
import java.util.concurrent.TimeUnit;

/**
 * 发送单向消息
 *
 * @author wanghuilong
 * @version 1.0
 * @date 2021/3/31 5:44 下午
 */
@Slf4j
public class OneWayProducer {

    public static void main(String[] args) throws MQClientException, NoSuchMethodException, InvocationTargetException
            , IllegalAccessException, InterruptedException {

        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("test_group1");
        // Specify name server addresses.
        producer.setNamesrvAddr("10.211.55.3:9876");
        producer.setSendMsgTimeout(10000);
        //Launch the instance.
        producer.start();

        try {
            for (int i = 0; i < 10; i++) {
                final int index = i;
                //Create a message instance, specifying topic, tag and message body.
                Message msg = new Message("TopicTest" /* Topic */,
                        "TagC" /* Tag */,
                        "KeyC", /*keys*/
                        ("Hello RocketMQ OneWay " +
                                i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                );
                // SendCallback接收异步返回结果的回调
                log.info("before send index:{}", index);
                producer.sendOneway(msg);
                TimeUnit.SECONDS.sleep(1);
            }
            //Shut down once the producer instance is not longer in use.
        } catch (Exception e) {
            e.printStackTrace();
        }
        producer.shutdown();

    }
}
